@thi.ng/rstream
Advanced tools
Reactive multi-tap streams, dataflow & transformation pipeline constructs
This project is part of the @thi.ng/umbrella monorepo.
Lightweight reactive multi-tap streams and transducer based transformation pipeline constructs, written in TypeScript.
This library provides & uses three key building blocks for reactive programming:
(No value judgements implied - there's room for both approaches!)
merge() /
sync())yarn add @thi.ng/rstream
Several demos in this repo's /examples directoy are using this package.
A small selection:
This demo is utilizing the @thi.ng/rstream-graph support package.
The FPS counter canvas component used in this benchmark is driven by this package.
Creates a new Stream instance, optionally with given StreamSource
function and / or ID. If a src function is provided, the function
will be only called (with the Stream instance as single argument)
once the first subscriber has attached to the stream. If the function
returns another function, it will be used for cleanup purposes if the
stream is cancelled, e.g. if the last subscriber has unsubscribed.
Streams are intended as (primarily async) data sources in a dataflow
graph and are the primary construct for the various from*()
functions provided by the package. However, streams can also be
triggered manually (from outside the stream), in which case the user
should call stream.next() to cause value propagation.
a = rs.stream((s) => {
s.next(1);
s.next(2);
s.done()
});
a.subscribe(trace("a"))
// a 1
// a 2
// a done
// as reactive value mechanism
b = rs.stream();
// or alternatively
// b = rs.subscription();
b.subscribe(trace("b1"));
b.subscribe(trace("b2"));
// external trigger
b.next(42);
// b1 42
// b2 42
Streams (like Subscriptions) implement the @thi.ng/api IDeref
interface which provides read access to a stream's last received value.
This is useful for various purposes, e.g. in combination with
@thi.ng/hdom, which supports direct embedding of streams (i.e. their
values) into UI components (and will be deref'd automatically). If the
stream has not yet emitted a value or if the stream is done, it will
deref to undefined.
Creates a new Subscription instance, the fundamental datatype &
building block provided by this package (Streams are Subscriptions
too). Subscriptions can be:
IDeref interface// as reactive value mechanism (same as with stream() above)
s = rs.subscription();
s.subscribe(rs.trace("s1"));
s.subscribe(rs.trace("s2"), tx.filter((x) => x > 25));
// external trigger
s.next(23);
// s1 23
s.next(42);
// s2 42
// s1 42
Returns a new StreamMerge instance, a subscription type consuming
inputs from multiple inputs and passing received values on to any
subscribers. Input streams can be added and removed dynamically. By
default, StreamMerge calls done() when the last active input is
done, but this behavior can be overridden via the close option (set it
to false).
merge({
// input streams w/ different frequencies
src: [
fromIterable([1, 2, 3], 10),
fromIterable([10, 20, 30], 21),
fromIterable([100, 200, 300], 7)
]
}).subscribe(trace());
// 100
// 1
// 200
// 10
// 2
// 300
// 3
// 20
// 30
Use the labeled()
transducer
for each input to create a stream of labeled values and track their
provenance:
merge({
src: [
fromIterable([1, 2, 3]).transform(labeled("a")),
fromIterable([10, 20, 30]).transform(labeled("b")),
]
}).subscribe(trace());
// ["a", 1]
// ["b", 10]
// ["a", 2]
// ["b", 20]
// ["a", 3]
// ["b", 30]
See StreamMergeOpts for further reference of the various behavior options.
Similar to StreamMerge above, but with extra synchronization of
inputs. Before emitting any new values, StreamSync collects values
until at least one has been received from all inputs. Once that's the
case, the collected values are sent as labeled tuple object to
downstream subscribers. Each value in the emitted tuple objects is
stored under their input stream's ID. Only the last value received from
each input is passed on. After the initial tuple has been emitted, you
can choose from two possible behaviors:
reset option is true, every input will have to provide at
least one new value again until another result tuple is produced.Any done inputs are automatically removed. By default, StreamSync
calls done() when the last active input is done, but this behavior can
be overridden via the close constructor option (set to false).
const a = rs.stream();
const b = rs.stream();
s = sync({ src: { a, b } }).subscribe(trace("result: "));
a.next(1);
b.next(2);
// result: { a: 1, b: 2 }
Input streams can be added and removed dynamically and the emitted tuple size adjusts to the current number of inputs (the next time a value is received from any input).
If the reset option is enabled, the last emitted tuple is allowed to
be incomplete, by default. To only allow complete tuples, also set the
all option to false.
The synchronization is done via the
partitionSync()
transducer from the @thi.ng/transducers package. See this function's
docs for further details.
See StreamSyncOpts for further reference of the various behavior options.
Topic based stream splitter. Applies topic function to each
received value and only forwards it to child subscriptions for
returned topic. The actual topic (return value from topic fn) can
be of any type, apart from undefined. Complex topics (e.g objects /
arrays) are allowed and they're matched with registered topics using
@thi.ng/equiv by default (but customizable via equiv option).
Each topic can have any number of subscribers.
If a transducer is specified for the PubSub, it is always applied
prior to passing the input to the topic function. I.e. in this case
the topic function will receive the transformed inputs.
PubSub supports dynamic topic subscriptions and unsubscriptions via
subscribeTopic() and unsubscribeTopic(). However, the standard
subscribe() / unsubscribe() methods are NOT supported (since
meaningless here) and will throw an error! unsubscribe() can only be
called WITHOUT argument to unsubscribe the entire PubSub instance
(incl. all topic subscriptions) from the parent stream.
Returns a new PubSub instance using given predicate pred as boolean
topic function and a & b as subscribers for truthy (a) and falsy
b values.
rs.fromIterable([1, 2, 3, 4]).subscribe(
rs.bisect(
(x) => !!(x & 1),
rs.trace("odd"),
rs.trace("even")
)
);
// odd 1
// even 2
// odd 3
// even 4
// odd done
// even done
If a or b need to be subscribed to directly, then a / b MUST be
first created as Subscription (if not already) and a reference kept
prior to calling bisect().
const odd = rs.subscription();
const even = rs.subscription();
odd.subscribe(rs.trace("odd"));
odd.subscribe(rs.trace("odd x10"), tx.map((x)=> x * 10));
even.subscribe(rs.trace("even"));
rs.fromIterable([1, 2, 3, 4]).subscribe(rs.bisect((x) => !!(x & 1), odd, even));
// odd x10 10
// odd 1
// even 2
// odd x10 30
// odd 3
// even 4
// odd done
// odd x10 done
// even done
Buffers values from src until side chain fires, then emits buffer
(unless empty) and repeats process until either input is done. By
default, the value read from the side chain is ignored, however the
optional predicate can be used to only trigger for specific values /
conditions.
// merge various event streams
merge([
fromEvent(document,"mousemove"),
fromEvent(document,"mousedown"),
fromEvent(document,"mouseup")
])
// queue event processing to only execute during the
// requestAnimationFrame cycle (RAF)
.subscribe(sidechainPartition(fromRAF()))
.subscribe(trace())
Filters values from input based on values received from side chain. By default, the value read from the side chain is ignored, however the optional predicate can be used to only trigger for specific values/conditions. Every time the predicate fn returns true, the filter will be toggled on/off. Whilst switched off, no input values will be forwarded.
// use slower interval stream to toggle main stream on/off
fromInterval(500)
.subscribe(sidechainToggle(fromInterval(1000)))
.subscribe(trace());
// 0
// 3
// 4
// 7
// 8
...
Delegate stream value processing to workers and pass on their responses to downstream subscriptions. Supports multiple worker instances and worker termination / restart for each new stream value received.
Send values to workers (incl. optional (inline) worker instantiation)
Create value stream from worker messages.
© 2017 - 2018 Karsten Schmidt // Apache Software License 2.0
FAQs
Reactive streams & subscription primitives for constructing dataflow graphs / pipelines
The npm package @thi.ng/rstream receives a total of 339 weekly downloads. As such, @thi.ng/rstream popularity was classified as not popular.
We found that @thi.ng/rstream demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 0 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Newly introduced telemetry in devenv 1.4 sparked a backlash over privacy concerns, leading to the removal of its AI-powered feature after strong community pushback.
Security News
TC39 met in Seattle and advanced 9 JavaScript proposals, including three to Stage 4, introducing new features and enhancements for a future ECMAScript release.
Security News
Deno 2.2 enhances Node.js compatibility, improves dependency management, adds OpenTelemetry support, and expands linting and task automation for developers.